package io.mqttpush.mqttserver.handles;

import io.mqttpush.common.events.MsgEvent;
import io.mqttpush.common.events.OffLineEvent;
import io.mqttpush.mqttserver.config.DeviceManager;
import io.mqttpush.mqttserver.msgs.MessageService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * 抽象的channelhandle
 * 提供了基本的操作
 *
 * @author tzj
 */
public class DispatchHandle extends ChannelInboundHandlerAdapter {


    protected Logger logger = LoggerFactory.getLogger(getClass());


    List<Iservice> iservices = new ArrayList<>();

    final Iservice loginService = new LoginService();

    final DeviceManager deviceManager = DeviceManager.getDeviceManager();



    static class  ProcessState{

        ProcessStateEnum state;

        public ProcessState(ProcessStateEnum state) {
            this.state = state;
        }

        public ProcessStateEnum getState() {
            return state;
        }

        public void setState(ProcessStateEnum state) {
            this.state = state;
        }
    }

   static  enum  ProcessStateEnum{

       normal,error;
   }


    public void init() {


        iservices.add(new PingService());
        iservices.add(new SubscribeService());
        iservices.add(new DisConnectService());
        iservices.add(new MessageService());

    }

    /**
     * 当有报文来的时候执行
     *
     * @param outMessage
     */
    public List<Object> service(ChannelHandlerContext context, List<Object> outMessage, MqttMessage inMessage) {



        Channel channel=context.channel();
        /**
         *
         * 未登录的必须先登录
         */
        if (!deviceManager.isLogin(channel)) {

            if (loginService.accept(context, inMessage)) {
                return loginService.service(context, outMessage, inMessage);
            } else {
                logger.info("have never login in {},{}", channel, inMessage);

                outMessage.stream().findFirst().ifPresent((state)->{
                    ((ProcessState)(state)).setState(ProcessStateEnum.error);
                });
            }

        } else {
            for (Iservice iservice : iservices) {
                if (iservice.accept(context, inMessage)) {
                    return iservice.service(context, outMessage, inMessage);
                }

            }
        }


        return outMessage;

    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {


        init();
        /**
         * 注册关闭事件通知
         */
        ctx.channel().closeFuture().addListener(l -> {

            if (l.isSuccess()) {


                Channel channel = ctx.channel();

                String deviceId = deviceManager.deviceId(channel);

                if (deviceId != null) {
                    /**
                     * 设备退出
                     */
                    deviceManager.deviveLoginOut(deviceId, channel);

                    String userInfo = deviceManager.userInfo(channel);

                    if (deviceId != null) {
                        OffLineEvent offLineEvent = new OffLineEvent();
                        offLineEvent.deviceId = deviceId;
                        offLineEvent.eventTime = new Date();
                        offLineEvent.userInfo = userInfo;
                        ctx.fireUserEventTriggered(offLineEvent);
                    }
                }


            }
        });
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {


        if (!(msg instanceof MqttMessage)) {
            ctx.close();
        }


        final List<Object> outMessage = new ArrayList<>();
        outMessage.add(new ProcessState(ProcessStateEnum.normal));

        /**
         * 在业务线程里面执行
         */
        CompletableFuture<List<Object>> completableFuture = CompletableFuture.supplyAsync(

                () -> this.service(ctx, outMessage, (MqttMessage) msg));


        /**
         * 在IO线程里面reduce
         */
        completableFuture.thenAccept((outMessages) -> {



            outMessage.stream().findFirst().ifPresent((state)->{

                if(((ProcessState)(state)).getState()==ProcessStateEnum.error){

                    logger.info("oops an errot,close the channel!");
                    ctx.close();
                    return;
                }
            });

            if (outMessage == null || outMessages.isEmpty()) {
                ctx.fireChannelRead(msg);
            } else {


                /**
                 *
                 * 返回值可能是mqtt消息和 事件
                 * 把消息写出去
                 *
                 * 把事件触发
                 */

                for (Object outObj : outMessages) {
                    if (outObj instanceof MqttMessage) {
                        ctx.write(outObj);
                    } else if (outObj instanceof MsgEvent) {
                        ctx.fireUserEventTriggered(outObj);
                    }
                }
                ctx.flush();
            }

        });


    }


}
